package com.cloudputing.hbase.spring.wiki.dao;

import java.util.List;
import java.util.UUID;

import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.RowMapper;
import org.springframework.data.hadoop.hbase.TableCallback;

import com.cloudputing.hbase.spring.wiki.valueobject.Wiki;

public class WikiDao {

	public static final byte[] CF_INFO = Bytes.toBytes("text");
	private HbaseTemplate hbaseTemplate; 
	private String tableName = "wiki";
	private byte[] qName = Bytes.toBytes("qual1");
	private byte[] qPhone = Bytes.toBytes("qual2");

	public HbaseTemplate getHbaseTemplate() {
		return hbaseTemplate;
	}

	public void setHbaseTemplate(HbaseTemplate hbaseTemplate) {
		this.hbaseTemplate = hbaseTemplate;
	}

	// constructor omitted
	public List<Wiki> findAll() {
		return hbaseTemplate.find(tableName, "text", new RowMapper<Wiki>() {
			@Override
			public Wiki mapRow(Result result, int rowNum) throws Exception {
				Wiki wiki = new Wiki();
				wiki.setName(Bytes.toString(result.getValue(CF_INFO, qName)));
				wiki.setPhone(Bytes.toString(result.getValue(CF_INFO, qPhone)));
				return wiki;
			}
		});
	}
	
	public Wiki save(final Wiki wiki) {
		return hbaseTemplate.execute(tableName, new TableCallback<Wiki>() {
			public Wiki doInTable(HTable table) throws Throwable {
				Put p = new Put(Bytes.toBytes(UUID.randomUUID().toString()));
				p.add(CF_INFO, qName, Bytes.toBytes(wiki.getName()));
				p.add(CF_INFO, qPhone, Bytes.toBytes(wiki.getPhone()));
				table.put(p);
				return wiki;
			}
		});
	}

}
